package org.luckyjourney.service.impl;

import io.jsonwebtoken.lang.Objects;
import org.luckyjourney.entity.AggregatedUpdate;
import org.luckyjourney.entity.vo.ModelUpdate;
import org.luckyjourney.exception.BaseException;
import org.luckyjourney.service.UpdateModelService;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * @creator: liRong
 * @date: 2025/6/26 14:50
 */
@Service
public class UpdateModelServiceImpl implements UpdateModelService {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    // 增量更新队列
    private final Queue<ModelUpdate> modelUpdateQueue = new ConcurrentLinkedQueue<>();

    // 聚合缓存
    private final Map<String , AggregatedUpdate> aggregationCache = new ConcurrentHashMap<>();

    // 批处理时间窗口 (毫秒)
    private static final long BATCH_WINDOW = 300;

    // 添加流量控制
    private static final int MAX_QUEUE_SIZE = 10000;


    @Override
    public void addUpdate(ModelUpdate updateModel) {
        if(modelUpdateQueue.size() < MAX_QUEUE_SIZE) {
            modelUpdateQueue.offer(updateModel);
        } else{
            throw new BaseException("当前流量过大！！服务出现故障");
        }
    }

    // 生成聚合键 (userId + label + videoId)
    private String getAggregationKey(ModelUpdate update) {
        return update.getUserId() + "|" +
                update.getModel().getLabel();
    }

    // 每300毫秒 触发批处理并发送到 RabbitMQ
    @Scheduled(fixedRate = BATCH_WINDOW)
    public void processAndSendBatchUpdates() {
        if(modelUpdateQueue.isEmpty()) return;

        // 1.处理队列中的所有更新
        while(!modelUpdateQueue.isEmpty()) {
            ModelUpdate update = modelUpdateQueue.poll();
            if(update == null) continue;

            String key = getAggregationKey(update);
            aggregationCache.compute(key , (k , aggregate) -> {
               if(aggregate == null) {
                    aggregate = new AggregatedUpdate();
                    aggregate.setUserId(update.getUserId());
                    aggregate.setLabel(update.getModel().getLabel());
                    aggregate.setVideoId(update.getModel().getVideoId());
               }
               aggregate.addScore(update.getModel().getScore());
               return aggregate;
            });
        }

        // 2.发送聚合结果到 RabbitMQ
        if(!aggregationCache.isEmpty()) {
            // 按用户ID分组聚合 结果
            Map<Long , List<AggregatedUpdate>> groupedUpdates = new HashMap<>();

            for (AggregatedUpdate aggregatedUpdate : aggregationCache.values()) {
                groupedUpdates.computeIfAbsent(aggregatedUpdate.getUserId() , k -> new ArrayList<>())
                        .add(aggregatedUpdate);
            }

            // 3. 发送到 RabbitMQ
            rabbitTemplate.convertAndSend(
                    "model.aggregation.exchange",
                    "model.aggregation.routingKey",
                    groupedUpdates
            );

            // 4. 清空缓存
            aggregationCache.clear();
        }

    }
}
